package org.eminem.hadoop.mapper;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.TableMapper;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.NullWritable;
import my.elasticsearch.action.bulk.BulkRequestBuilder;
import my.elasticsearch.action.bulk.BulkResponse;
import my.elasticsearch.action.update.UpdateRequestBuilder;
import my.elasticsearch.client.Client;

import org.eminem.elasticsearch.ESClient;
import org.eminem.hbase.HbaseClient;

import com.alibaba.fastjson.JSONArray;

/**
 * ES init mapper
 */
public class ESInitMapper extends TableMapper<NullWritable, NullWritable> {

	private Client client;
	private static final Log LOG = LogFactory.getLog(ESInitMapper.class);
	private BulkRequestBuilder bulkRequestBuilder;
	private String index;
	private String type;
	
	private HTable table ;

	/**
	 * init ES
	 * @param context
	 * @throws IOException
	 * @throws InterruptedException
	 */
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		try {
			Configuration conf = context.getConfiguration() ;
			String clusterName = conf.get("es.cluster.name");
			String host = conf.get("es.cluster.host");
			int port = Integer.parseInt(conf.get("es.cluster.port"));
			table = HbaseClient.getHTable(conf, table, conf.get("hbase.qxtable.name")) ;
			client = ESClient.getEsClient(clusterName,host,port);
			bulkRequestBuilder = client.prepareBulk();
			bulkRequestBuilder.setRefresh(true);
			index = conf.get("index");
			type = conf.get("type");
		} catch (Exception ex) {
			LOG.info("初始化MAP异常:" + ex.getMessage());
		}
	}

	@Override
	protected void map(ImmutableBytesWritable key, Result values, Context context) throws IOException, InterruptedException {
		Map<String, Object> infoJson = new HashMap<String, Object>();
		Map<String,Object> familyMap = new HashMap<String,Object>() ;
		
		for(Cell cell : values.rawCells()) {
			//获取标签的时间
			String time = timeStamp2Date(cell.getTimestamp()) ;
			String timeStr = timeStamp2DateStr(cell.getTimestamp()) ;
			//获取标签的值
			//JSONArray jsonArray = JSONArray.parseArray(Bytes.toString(CellUtil.cloneValue(cell))) ;
			JSONArray jsonArray = JSONArray.parseArray(Bytes.toString(CellUtil.cloneValue(cell))) ;
			//获取标签的列
			String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell)) ;
			Map<String,Object> qualifierMap = new HashMap<String,Object>() ;
			if(qualifierMap.containsKey(qualifier)) {
				List<Map<String,String>> list = (List<Map<String, String>>) qualifierMap.get(qualifier) ;
				for (Object obj : jsonArray) {
					Map<String,String> map = new HashMap<String,String>() ;
					map.put("label", obj.toString()) ;
					list.add(map) ;
				}
			}else {
				List<Map<String,String>> list = new ArrayList<Map<String,String>>() ;
				for (Object obj : jsonArray) {
					Map<String,String> map = new HashMap<String,String>() ;
					map.put("label", obj.toString()) ;
					list.add(map) ;
				}
				qualifierMap.put(qualifier, list) ;
			}
			//获取到列簇
			String family = Bytes.toString(CellUtil.cloneFamily(cell)) ;
			if(familyMap.containsKey(family)) {
				Set<Map<String,Object>> set = (Set<Map<String, Object>>) familyMap.get(family) ;
				set.add(qualifierMap) ;
			}else {
				Set<Map<String,Object>> set = new HashSet<Map<String,Object>>() ;
				set.add(qualifierMap) ;
				familyMap.put(family, set) ;
				Map<String,Object> timeMap = new HashMap<String,Object>() ;
				timeMap.put("label", time) ;
				Map<String,Object> nyMap = new HashMap<String,Object>() ;
				nyMap.put("label", timeStr) ;
				Map<String,Object> mapTime = new HashMap<String,Object>() ;
				Map<String,Object> mapNy = new HashMap<String,Object>() ;
				mapTime.put("time", timeMap) ;
				mapNy.put("ny", nyMap) ;
				set.add(mapTime) ;
				set.add(mapNy) ;
			}
		}
		List<Object> list = new ArrayList<Object>() ;
		for (Entry<String, Object> entry : familyMap.entrySet()) {
			Object value = entry.getValue();
			list.add(value) ;
		}
		infoJson.put("history", list);
		Get get = new Get(values.getRow()) ;
		get.addColumn(Bytes.toBytes("ids"), Bytes.toBytes("superid")) ;
		Result result = table.get(get) ;
		if(!result.isEmpty()) {
			String id = Bytes.toString(result.value()) ;
			addUpdateBuilderToBulk(client.prepareUpdate(index, type, id).setDocAsUpsert(true).setDoc(infoJson));
		}
	}

	@Override
	protected void cleanup(Context context) throws IOException, InterruptedException {
		if (0 < bulkRequestBuilder.numberOfActions()) {
			try {
				bulkRequest();
			} catch (Exception ex) {
				LOG.info("Bulk " + index + " index error :" + ex.getMessage());
			}
		}
		//关闭es
		HbaseClient.closeHTable(table) ;
		client.close();
	}

	/**
	 * execute bulk process
	 * @throws Exception
	 */
	private void bulkRequest() throws Exception {
		BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();
		if (!bulkResponse.hasFailures()) {
			bulkRequestBuilder = client.prepareBulk();
		}
	}

	/**
	 *add prepare update date  to builder
	 * @param builder
	 */
	private synchronized void addUpdateBuilderToBulk(UpdateRequestBuilder builder) {
		try {
			if (bulkRequestBuilder.numberOfActions() != 0 && (bulkRequestBuilder.numberOfActions() % 500 == 0)) {
				bulkRequest();
			}
			bulkRequestBuilder.add(builder);
		} catch (Exception ex) {
			LOG.info("add Bulk " + index + " index error :" + ex.getMessage());
		}
	}

	private static String timeStamp2Date(Long seconds) {
		SimpleDateFormat sdf = new SimpleDateFormat("yyyyMMdd");
		return sdf.format(new Date(seconds));
	}
	
	private static String timeStamp2DateStr(Long seconds) {
		SimpleDateFormat sdf = new SimpleDateFormat("yyyy年MM月");
		return sdf.format(new Date(seconds));
	}
	
}
